package com.demo.webflux.feign;

import java.time.Duration;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.demo.webflux.Constant;
import com.demo.webflux.PropertiesInnerUtil;

import dev.miku.r2dbc.mysql.MySqlConnectionConfiguration;
import dev.miku.r2dbc.mysql.MySqlConnectionFactory;
import io.r2dbc.spi.Connection;
import io.r2dbc.spi.ConnectionFactory;
import reactivefeign.webclient.WebReactiveFeign;
import reactor.core.publisher.Mono;
import reactor.netty.DisposableServer;
import reactor.netty.http.server.HttpServer;

public class Demo1 {
	private static Logger logger = LoggerFactory.getLogger(Demo1.class);
	private static String userName = PropertiesInnerUtil.getString(Constant.DATASOURCE_LOCATION, "datasource.username");
	private static String password = PropertiesInnerUtil.getString(Constant.DATASOURCE_LOCATION, "datasource.password");

	public static void main(String[] args) {
		MySqlConnectionConfiguration.Builder builder = MySqlConnectionConfiguration.builder()
	            .host("10.16.9.34")
	            .port(3306)
	            .connectTimeout(Duration.ofSeconds(3))
	            .username(userName)
	            .password(password)
	            .database("sms_crm_db");

		MySqlConnectionConfiguration configuration = builder.build();
		ConnectionFactory connectionFactory = MySqlConnectionFactory.from(configuration);
		Mono<Connection> connectionMono = Mono.from(connectionFactory.create());
		FeignClientService feignClientService = WebReactiveFeign  //WebClient based reactive feign
	            .<FeignClientService>builder()
	            .target(FeignClientService.class, "http://127.0.0.1:8085");

        DisposableServer server = HttpServer.create()
				            		.port(8085)
				            		.route(routes ->
				            			routes
				        				// 当我们发出一个GET请求去访问/hello时就会得到一个字符串Hello World!
				        				.get("/feign", (request, response) -> {
				        					logger.info("main");
				        			    	Mono<String> result = connectionMono.flatMapMany(connection -> {
				        			    		logger.info("select");
				        						return connection.createStatement("SELECT cid, logic_code, message_send_code, message_content FROM message_queue_receiver limit 10000").execute();
				        					})
				        			    	.flatMap(rs -> {
				        			    		logger.info("flatMap");
				        			    		return rs.map((row, metadata) -> row.get(1, String.class));
				        			    	})
				        			    	.collectList()
				        			    	.flatMap(list -> {
				        			    		Object data = list.get(0);
				        			    		return feignClientService.test(data + "");
				        			    	});
				        			    	logger.info("main end");
				        					return response.sendString(result);
				        				})
				        				.get("/log/{name}", (request, response) -> {
				        					try {
				        						Thread.sleep(3000);
				        					} catch (InterruptedException e) {
				        						e.printStackTrace();
				        					}
				        					return response.sendString(Mono.just("hello: " + request.param("name")));
				        				}))
				            		.bindNow();

        server.onDispose()
              .block();
    }

}
